package com.flute.haflute.jobbox.child;

import java.rmi.Remote;
import java.rmi.RemoteException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.jobbox.base.CloudJobConf;
import com.flute.haflute.jobbox.base.CloudTask;
import com.flute.haflute.jobbox.base.InputResource;
import com.flute.haflute.jobbox.common.CloudConstants;
import com.flute.haflute.jobbox.common.CloudTaskReporter;
import com.flute.haflute.jobbox.communication.ChildCommunicationService;
import com.flute.haflute.jobbox.communication.messages.CloudJobReqMessage;
import com.flute.haflute.jobbox.communication.messages.MessageBase;

/**
 * 非MapReduce任务调度器， 任务环境初始化工作 任务初始化 任务配置 任务执行
 */
public class CloudTaskScheduler implements TaskScheduler {
	private static Logger logger = LoggerFactory
			.getLogger(CloudTaskScheduler.class);

	/** 子节点通信平台的接口 */
	private ChildCommunicationService communicationService;

	public CloudTaskScheduler(ChildCommunicationService service) {
		communicationService = service;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.cpkf.yyjd.cloud.child.TaskScheduler#scheduleTask(com.cpkf.yyjd.cloud
	 * .communication.messages.RequestMessage)
	 */
	@Override
	public void scheduleTask(MessageBase request) throws Exception {
		executeCloudTask((CloudJobReqMessage) request);
	}

	/** 真正调度一个非MapReduce任务 */
	private void executeCloudTask(CloudJobReqMessage req)
			throws RemoteException, InstantiationException,
			IllegalAccessException {
		CloudJobConf jobConf = (CloudJobConf) req.getContent();
		logger.info("Child <{}> start to run one TASK: {} ",
				ChildContext.getDaemonNo(), jobConf.toString());
		/*
		 * //set IO environment Properties props = new Properties(); String[]
		 * files = jobConf.getTransferFiles(); if (files != null && files.length
		 * > 0) { for (String file : files) { if(file.endsWith(".jar") ||
		 * file.endsWith(".class")) { //load jar or class } else { File
		 * configFile = new File(file); if(configFile.exists()) {
		 * props.setProperty(configFile.getName(),
		 * configFile.getAbsolutePath()); } } } }
		 */
		Class<? extends CloudTask> runClazz = jobConf.getTaskRunner();
		// 实例化runner类
		CloudTask runner = runClazz.newInstance();

		String sId = req.getSourceId();
		Remote inputResource = null;
		if (!sId.equals(CloudConstants.JOB_SOURCE_INPUT_NONE)) {
			inputResource = communicationService.getMessager(sId);
		}

		sId = req.getMessagerId();
		Remote messager = null;
		if (!sId.equals(CloudConstants.JOB_REQ_MESSAGER_NONE)) {
			messager = communicationService.getMessager(sId);
		}

		if (inputResource != null) {
			// 设置输入资源
			runner.setInput((InputResource) inputResource);
		}

		if (messager != null) {
			// 设置中心控制器
			runner.configJobAgent(messager);
		}
		runner.configureTask(jobConf.getJobContext());
		// 启动Runner
		runner.start(new CloudTaskReporter());
		logger.info("Child <{}> started TASK<{}>, it will run backend",
				ChildContext.getDaemonNo(), jobConf.getJobId());
	}

}
